#include "thread_msg.h"

/* FrameCur */
int dev_read_msg_pub(uint8_t idx)
{
    cJSON *root, *js_ack, *js_dev, *js_array_item, *js_points, *js_key;
    uint16_t frameCur[] = {1600, 2000, 2500, 3200, 4000, 6300};

    root = cJSON_CreateObject();
    PARAM_CHECK_IS_VALID(!root, "can't create cjson object", -1);

    cJSON_AddNumberToObject(root, "tick", 1);
    cJSON_AddNumberToObject(root, "time", 1234);
    cJSON_AddStringToObject(root, "src", "dev_modbus_usb");

    js_dev = cJSON_CreateObject();
    PARAM_CHECK_IS_VALID_AND_EXEC_CMD(!js_dev, "can't create dev object",
                                      cJSON_Delete(root), -1);

    cJSON_AddStringToObject(js_dev, "status", "online");

    js_points = cJSON_CreateArray();
    PARAM_CHECK_IS_VALID_AND_EXEC_CMD(!js_points, "can't create points object",
                                      cJSON_Delete(root), -1);

    js_key = cJSON_CreateObject();
    cJSON_AddStringToObject(js_key, "key", "FrameCur");
    cJSON_AddNumberToObject(js_key, "val", frameCur[idx]);
    cJSON_AddItemToArray(js_points, js_key);
    cJSON_AddItemToObject(js_dev, "points", js_points);
    cJSON_AddItemToObject(root, "dev", js_dev);

    char *newpayload = cJSON_PrintUnformatted(root);
    PARAM_CHECK_IS_VALID_AND_EXEC_CMD(!newpayload, "malloc error",
                                      cJSON_Delete(root), -1);
    /* get topic */
    string new_topic ="dev/dev_modbus_usb/0/info";                            
    /* publish */
    devModbus->mqtt->Publish(new_topic.c_str(), newpayload);
    cJSON_free(newpayload);
    cJSON_Delete(root);
    return 0;
}

int dev_read_485_msg_pub(uint16_t param)
{
    cJSON *root, *js_ack, *js_dev, *js_array_item, *js_points, *js_key;
    //uint16_t frameCur[] = {1600, 2000, 2500, 3200, 4000, 6300};

    root = cJSON_CreateObject();
    PARAM_CHECK_IS_VALID(!root, "can't create cjson object", -1);

    cJSON_AddNumberToObject(root, "tick", 1);
    cJSON_AddNumberToObject(root, "time", 1234);
    cJSON_AddStringToObject(root, "src", "dev_modbus_usb");

    js_dev = cJSON_CreateObject();
    PARAM_CHECK_IS_VALID_AND_EXEC_CMD(!js_dev, "can't create dev object",
                                      cJSON_Delete(root), -1);

    cJSON_AddStringToObject(js_dev, "status", "online");

    js_points = cJSON_CreateArray();
    PARAM_CHECK_IS_VALID_AND_EXEC_CMD(!js_points, "can't create points object",
                                      cJSON_Delete(root), -1);

    js_key = cJSON_CreateObject();
    cJSON_AddStringToObject(js_key, "key", "FrameCur");
    cJSON_AddNumberToObject(js_key, "val", param);
    cJSON_AddItemToArray(js_points, js_key);
    cJSON_AddItemToObject(js_dev, "points", js_points);
    cJSON_AddItemToObject(root, "dev", js_dev);

    char *newpayload = cJSON_PrintUnformatted(root);
    PARAM_CHECK_IS_VALID_AND_EXEC_CMD(!newpayload, "malloc error",
                                      cJSON_Delete(root), -1);
    /* get topic */
    string new_topic ="dev/dev_modbus_usb/0/info";                            
    /* publish */
    devModbus->mqtt->Publish(new_topic.c_str(), newpayload);
    cJSON_free(newpayload);
    cJSON_Delete(root);
    return 0;
}

int dev_read_msg_deal(string topic, string payload)
{
    return 0;
}

int dev_write_msg_deal(string topic, string payload)
{
    return 0;
}

/* thread-----------------thread------------------thread------------thread */
void dev_auto_read_queue_deal_thread(void)
{
    int ret;
    while (true) {
        modbus_t *ctx;
        ret = 0;
        uint16_t value = 0;
        ctx = modbus_new_rtu("/dev/ttyS3", 9600, 'E', 8, 1);
        modbus_set_debug(ctx, TRUE);
        if (modbus_connect(ctx) == -1) {
            sys_log_error("modbus ttyS3 Connect failed: %s\n", modbus_strerror(errno));
            modbus_free(ctx);
            usleep(1 * 1000 * 1000);
            continue;
        }
        modbus_set_slave(ctx, 0x01);
        modbus_set_response_timeout(ctx, 0, 999999);
        ret = modbus_read_registers(ctx, 0x0180, 1, &value);
        if (-1 == ret) {
            sys_log_error("modbus ttyS3 read failed: %s\n", modbus_strerror(errno));
            modbus_close(ctx);
            modbus_free(ctx);
            usleep(1 * 1000 * 1000);
            continue;
        }
        modbus_close(ctx);
        modbus_free(ctx);
        sys_log_error("value: %d\n", value);
        dev_read_485_msg_pub(value);
        
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    }
}

/* mqtt消息处理线程 */
void dev_read_queue_deal_thread(void)
{
    msg_t msg;
    while (true) {
        if (!devModbus->dev_read.empty()) {
            msg = devModbus->dev_read.front();
            devModbus->dev_read.pop();
            if (-1 != msg.topic.find("/dev/cloud_chint/modbus_usb/read")) {
                dev_read_msg_deal(msg.topic, msg.payload);
            }
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    }
}

/* mqtt消息处理线程 */
void dev_write_queue_deal_thread(void)
{
    msg_t msg;
    while (true) {
        if (!devModbus->dev_write.empty()) {
            msg = devModbus->dev_write.front();
            devModbus->dev_write.pop();
            if (-1 != msg.topic.find("/dev/cloud_chint/modbus_usb/write")) {
                dev_write_msg_deal(msg.topic, msg.payload);
            }
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
}


/* mqtt消息处理线程 */
void mqtt_msg_deal_thread(void)
{
    msg_t msg;
    while (true) {
        if (!MqttMsgQueueIsEmpty()) {
            msg = MqttMsgQueueGetItem();
            sys_log_info("mqtt msg queue deal thread = %s:%s\n", msg.topic, msg.payload);
            MqttMsgQueuePop();
            string str = msg.topic;
            if (-1 != str.find("dev/") && -1 != str.find("/read")) {
                devModbus->dev_read.push(msg);
            } else if (-1 != str.find("dev/") && -1 != str.find("/write")) {
                devModbus->dev_write.push(msg);
            }
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(10)); /* 10ms 轮询一次mqtt队列 */
    }
}



